package com.spi.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageClientIDSetter;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;

@RocketMQMessageListener(
        topic = "${rocketmq.consumer.group3.topic}",
        consumerGroup = "${rocketmq.consumer.group3.consumer-group}",
        messageModel = MessageModel.CLUSTERING,
        selectorExpression = "tag1",
        selectorType = SelectorType.TAG
//        enableMsgTrace = true
)
@Component
@Slf4j
public class ConsumerGroup3ListenerDemo implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    @Override
    public void onMessage(MessageExt messageExt) {
        //tag方式过滤
        byte[] bytes = messageExt.getBody();
        String string = new String(bytes, StandardCharsets.UTF_8);
        System.out.println("111111111111111111111111111111111111111111");
        //注：messageExt.getMsgId()是个大坑，debug页面看messageExt里面的属性MsgId是与生产者的offsetMsgId一样，然后调用messageExt.getMsgId()得到的确实生产者的msgid
        //因为其内部加了转换逻辑
//        public String getMsgId() {
//            String uniqID = MessageClientIDSetter.getUniqID(this);
//            return uniqID == null ? this.getOffsetMsgId() : uniqID;
//        }
        log.info("消费者组3,标签为tag1下消费者，对应的消费者拿到的消息id为："+messageExt.getMsgId()+" 进行消费：" + string);
        //System.out.println("消费者组3,标签为tag1下消费者，对应的消费者拿到的消息id为："+messageExt.getMsgId()+" 进行消费：" + string);
        //throw new RuntimeException("消费者组3手动抛异常");
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.setMaxReconsumeTimes(3); //最大重试次数
        defaultMQPushConsumer.setInstanceName("instance2");
    }
}
